/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.datadispatch.handler.datadispatch;

import cn.ac.iie.ulss.datadispatch.commons.RuntimeEnv;
import com.alibaba.rocketmq.common.message.Message;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 *
 * @author ulss
 */
public class DataDispatcher {

    //doc schema
    Protocol protocol = null;
    String docSchemaName = null;
    String docSchemaContent = null;
    Schema docSchema = null;
    DatumReader<GenericRecord> docReader = null;
    //mq
    public String docStoreMQName = null;
    private BlockingQueue<Message> msgCache = null;
    private volatile boolean stopped;
    AtomicLong dataVolume = new AtomicLong();
    static Logger logger = null;

    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = Logger.getLogger(DataDispatcher.class.getName());
    }

    private DataDispatcher(String pDocSchemaName, String pDocSchemaContent, String pDocStoreMQName) {
        docSchemaName = pDocSchemaName;
        docSchemaContent = pDocSchemaContent;
//        protocol = Protocol.parse(docSchemaContent);//maybe failed
//        docSchema = protocol.getType(docSchemaName);
//        docReader = new GenericDatumReader<>(docSchema);
        docStoreMQName = pDocStoreMQName;
        msgCache = new ArrayBlockingQueue<Message>((Integer) RuntimeEnv.getParam(RuntimeEnv.MSG_CACHE_SIZE));
        stopped = false;
        for (int i = 0; i < (Integer) RuntimeEnv.getParam(RuntimeEnv.MSG_DW_NUM); i++) {
            DispatcherWorker dispatcherWorker = new DispatcherWorker();
            dispatcherWorker.start();
        }
    }

    public static DataDispatcher getDataDispatcher(String pDocSchemaName, String pDocSchemaContent, String pDocStoreMQName) {
        return new DataDispatcher(pDocSchemaName, pDocSchemaContent, pDocStoreMQName);

    }

    public void dispatch(String objID, byte[] pData) throws Exception {
        Message msg = new Message(docStoreMQName, pData);
        int tryTimes = 0;
        while (true) {
            if (msgCache.offer(msg)) {
                logger.debug(objID + ":msg cache for " + docSchemaName + " used " + msgCache.size() + ":" + tryTimes);
                break;
            } else {
                if ((++tryTimes) >= 10) {
                    throw new Exception("put msg into msg cache failed at last");
                }
            }
        }
    }

    class DispatcherWorker extends Thread {

        @Override
        public void run() {
            logger.info("dispatcher worker for " + docSchemaName + " " + Thread.currentThread().getId() + " starts to run ");
            Message msg = null;
            while (true) {
                try {
                    msg = msgCache.poll();
                    if (msg != null) {
                        long startTime = System.nanoTime();
                        //SendResult sr = RuntimeEnv.producer.send(msg);
                        long endTime = System.nanoTime();
                        //logger.debug("send " + docSchemaName + " message to " + sr.getMessageQueue().getBrokerName() + ":" + sr.getQueueOffset() + " ok:" + (endTime - startTime) / (1000 * 1000) + " status:" + sr.getSendStatus());
                        logger.debug("send " + docSchemaName + " message to " + " ok:" + (endTime - startTime) / (1000 * 1000));
                        msg = null;
                    } else {
                        try {
                            Thread.sleep(100);
                        } catch (Exception ex) {
                        }
                    }
                } catch (Exception ex) {
                    logger.warn("error happened when sending " + docSchemaName + " messge " + ex.getMessage(), ex);
                    while (true) {
                        if (msgCache.offer(msg)) {
                            logger.debug("put failed msg into msg cache successfully");
                            break;
                        } else {
                            try {
                                Thread.sleep(100);
                            } catch (Exception ex2) {
                            }
                        }
                    }
                }

                if (stopped) {
                    int msgCacheSize = msgCache.size();
                    if (msgCacheSize <= 0) {
                        break;
                    } else {
                        logger.debug("waiting " + msgCacheSize + " msg of " + docSchemaName + " to be cleared...");
                    }
                }
            }
            logger.debug("dispatcher for " + docSchemaName + " exits");
        }
    }

    public void shutdown() {
        stopped = true;
    }

    public String getDocSchemaContent() {
        return docSchemaContent;
    }

    public void incDataVolumeStatistics(long pDelta) {
        dataVolume.addAndGet(pDelta);
    }

    public long getDataVolumeStatistics() {
        return dataVolume.longValue();
    }
}
